package cn.yj.netty.test;

import cn.yj.netty.common.Const;
import cn.yj.netty.support.test.MessageProtocol;
import cn.yj.netty.common.CommandCode;
import cn.yj.netty.common.MessageType;
import cn.yj.netty.support.test.protocol.Header;
import com.alibaba.fastjson.JSON;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class NettyClient {

    private static final Logger log = LoggerFactory.getLogger(NettyClient.class);

    private static ChannelFuture channelFuture;
    private static EventLoopGroup group = new NioEventLoopGroup();

    private static Bootstrap bootstrap;

    private static void connect() {
        bootstrap =
                new Bootstrap().group(group).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                // 心跳监测机制
                socketChannel.pipeline().addLast(new IdleStateHandler(0, Const.INTERVAL_TIME, 0, TimeUnit.SECONDS));
                // 编码处理器
                socketChannel.pipeline().addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
                // 解码处理器
                socketChannel.pipeline().addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
                // 消息处理器
                socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {

                    /**
                     * 接收服务端信息
                     * @param ctx
                     * @param msg
                     * @throws Exception
                     */
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        System.out.println(msg.toString());
                        if (msg.toString().contains("messageType:PING")) {
                            MessageProtocol messageProtocol = MessageProtocol.newMessageProtocolHearPack();
                            messageProtocol.getHeader().setMessageType(MessageType.HEART_BEAT_PONG.getCode());
                            ctx.channel().writeAndFlush(messageProtocol.toProtocolStr());
                        }
                    }

                    /**
                     * 心跳触发
                     * @param ctx
                     * @param evt
                     * @throws Exception
                     */
                    @Override
                    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                        super.userEventTriggered(ctx, evt);
                    }

                    /**
                     * 连接断开触发
                     * @param ctx
                     * @throws Exception
                     */
                    @Override
                    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
                        ctx.channel().eventLoop().schedule(() -> {
                            log.warn("重新链接服务器1");
                            retryConnect();
                        }, 3, TimeUnit.SECONDS);
                    }
                });
            }
        });
        retryConnect();
    }

    private static synchronized void retryConnect() {
        channelFuture = bootstrap.connect("127.0.0.1", 8090);
    }

    public static void sendMsg(MessageProtocol protocol) throws InterruptedException {
        Channel channel = channelFuture.channel();
        ChannelFuture channelFuture = channel.writeAndFlush(protocol.toProtocolStr());
        if (channelFuture.await().isSuccess()) {
            log.info("发送成功");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        connect();
        MessageProtocol messageProtocol = new MessageProtocol();
        Header header = new Header();
        header.setVersion("1.0");
        header.setMessageType(MessageType.TEST.getCode());
        header.setCommandCode(CommandCode.TEST.getCode());
        header.setSessionId(UUID.randomUUID().toString());
        header.setTimestamp(new Date().getTime());
        header.setHost("127.0.0.1");
        messageProtocol.setHeader(header);
        HashMap<String, Object> stringObjectHashMap = new HashMap<>();
        stringObjectHashMap.put("code", 200);
        stringObjectHashMap.put("msg", "获取数据");
        messageProtocol.setBody(JSON.toJSONString(stringObjectHashMap));

        //NettyClient.sendMsg(messageProtocol);

        while (true) {
            messageProtocol.setHeader(messageProtocol.getHeader().setTimestamp(System.currentTimeMillis()).setSessionId(UUID.randomUUID().toString()));
            NettyClient.sendMsg(messageProtocol);
            Thread.sleep(35 * 1000);
        }
    }

}
